//! # 消息总线核心接口完整单元测试
//!
//! 为 MessageBus, Publisher, Subscriber 核心接口提供完整的单元测试覆盖
//! 确保每个函数、每个参数、每个边界条件都得到测试

use rustcloud_bus::prelude::*;
use rustcloud_bus::{
    bus::{MessageBus},
    memory::{InMemoryMessageBus},
    message::{Message, MessageType, MessagePriority},
    config::MessageBusConfig,
};
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use uuid::Uuid;

/// 测试辅助工具
struct TestHelper;

impl TestHelper {
    /// 创建测试消息总线
    async fn create_test_bus() -> Arc<InMemoryMessageBus> {
        Arc::new(InMemoryMessageBus::new().await.unwrap())
    }

    /// 创建带配置的测试消息总线
    async fn create_test_bus_with_config(config: MessageBusConfig) -> Arc<InMemoryMessageBus> {
        Arc::new(InMemoryMessageBus::with_config(config).await.unwrap())
    }

    /// 创建测试消息
    fn create_test_message(topic: &str, payload: serde_json::Value) -> Message {
        Message::new(topic, payload)
    }

    /// 创建不同类型的测试消息
    fn create_typed_message(topic: &str, msg_type: MessageType, payload: serde_json::Value) -> Message {
        match msg_type {
            MessageType::Event => Message::event(topic, payload),
            MessageType::Command => Message::command(topic, payload),
            MessageType::Query => Message::query(topic, payload),
            MessageType::Notification => Message::notification(topic, payload),
            MessageType::Reply => Message::reply(topic, payload, Uuid::new_v4()),
        }
    }

    /// 等待条件满足
    async fn wait_for_condition<F>(condition: F, timeout_ms: u64) -> bool 
    where
        F: Fn() -> bool + Send + 'static,
    {
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while start.elapsed() < timeout_duration {
            if condition() {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        false
    }
}

mod messagebus_interface_tests {
    use super::*;

    /// 测试 MessageBus::publish 方法的所有参数情况
    #[tokio::test]
    async fn test_publish_with_all_message_types() {
        let bus = TestHelper::create_test_bus().await;

        // 测试所有消息类型
        let message_types = vec![
            MessageType::Event,
            MessageType::Command,
            MessageType::Query,
            MessageType::Notification,
            MessageType::Reply,
        ];

        for msg_type in message_types {
            let message = TestHelper::create_typed_message(
                &format!("test.{:?}", msg_type).to_lowercase(),
                msg_type.clone(),
                json!({"type": format!("{:?}", msg_type)})
            );

            let result = bus.publish(message).await;
            assert!(result.is_ok(), "发布 {:?} 类型消息失败: {:?}", msg_type, result);
        }
    }

    /// 测试 MessageBus::publish 方法的边界条件
    #[tokio::test]
    async fn test_publish_boundary_conditions() {
        let bus = TestHelper::create_test_bus().await;

        // 测试空主题
        let empty_topic_msg = Message::new("", json!({}));
        let result = bus.publish(empty_topic_msg).await;
        assert!(result.is_err(), "空主题应该发布失败");

        // 测试非常长的主题
        let long_topic = "a".repeat(1000);
        let long_topic_msg = Message::new(&long_topic, json!({}));
        let result = bus.publish(long_topic_msg).await;
        assert!(result.is_ok(), "长主题应该发布成功");

        // 测试特殊字符主题
        let special_topic = "test.特殊字符.🎉.symbols";
        let special_msg = Message::new(special_topic, json!({"test": "special"}));
        let result = bus.publish(special_msg).await;
        assert!(result.is_ok(), "特殊字符主题应该发布成功");

        // 测试空载荷
        let empty_payload_msg = Message::new("test.empty", json!(null));
        let result = bus.publish(empty_payload_msg).await;
        assert!(result.is_ok(), "空载荷应该发布成功");

        // 测试大载荷
        let large_payload = json!({
            "data": "x".repeat(100_000), // 100KB
            "numbers": (0..1000).collect::<Vec<i32>>()
        });
        let large_msg = Message::new("test.large", large_payload);
        let result = bus.publish(large_msg).await;
        assert!(result.is_ok(), "大载荷应该发布成功");
    }

    /// 测试 MessageBus::publish 错误情况
    #[tokio::test]
    async fn test_publish_error_conditions() {
        let bus = TestHelper::create_test_bus().await;

        // 先关闭总线
        bus.close().await.unwrap();

        // 在关闭的总线上发布消息应该失败
        let message = TestHelper::create_test_message("test.closed", json!({}));
        let result = bus.publish(message).await;
        assert!(result.is_err(), "在关闭的总线上发布应该失败");
    }

    /// 测试 MessageBus::subscribe 方法的所有参数情况
    #[tokio::test]
    async fn test_subscribe_with_various_patterns() {
        let bus = TestHelper::create_test_bus().await;

        // 测试不同的订阅模式
        let test_patterns = vec![
            "exact.topic",           // 精确匹配
            "wildcard.*",           // 单级通配符
            "multilevel.**",        // 多级通配符
            "user.*.events",        // 中间通配符
            "user.events.*.*",      // 多个单级通配符
            "**",                   // 全匹配
            "a.b.c.d.e.f.g",       // 多级主题
            "数字.123.中文",         // 多语言字符
            "symbols.#.$.%",        // 特殊符号
        ];

        for pattern in test_patterns {
            let result = bus.subscribe(pattern).await;
            assert!(result.is_ok(), "订阅模式 '{}' 应该成功", pattern);
            
            let subscriber = result.unwrap();
            assert_eq!(subscriber.pattern(), pattern);
            assert!(subscriber.is_active());
            assert!(!subscriber.subscription_id().is_empty());
        }
    }

    /// 测试 MessageBus::subscribe 边界条件
    #[tokio::test]
    async fn test_subscribe_boundary_conditions() {
        let bus = TestHelper::create_test_bus().await;

        // 测试空模式
        let result = bus.subscribe("").await;
        assert!(result.is_err(), "空订阅模式应该失败");

        // 测试非常长的模式
        let long_pattern = "a.".repeat(500) + "*";
        let result = bus.subscribe(&long_pattern).await;
        assert!(result.is_ok(), "长订阅模式应该成功");

        // 测试重复订阅相同模式
        let pattern = "test.duplicate.subscription";
        let sub1 = bus.subscribe(pattern).await.unwrap();
        let sub2 = bus.subscribe(pattern).await.unwrap();
        
        // 应该创建不同的订阅ID
        assert_ne!(sub1.subscription_id(), sub2.subscription_id());
    }

    /// 测试 MessageBus::unsubscribe 方法
    #[tokio::test]
    async fn test_unsubscribe_all_cases() {
        let bus = TestHelper::create_test_bus().await;

        // 创建订阅
        let subscriber = bus.subscribe("test.unsubscribe").await.unwrap();
        let subscription_id = subscriber.subscription_id().to_string();

        // 正常取消订阅
        let result = bus.unsubscribe(&subscription_id).await;
        assert!(result.is_ok(), "正常取消订阅应该成功");

        // 重复取消订阅
        let result = bus.unsubscribe(&subscription_id).await;
        assert!(result.is_err(), "重复取消订阅应该失败");

        // 取消不存在的订阅
        let fake_id = Uuid::new_v4().to_string();
        let result = bus.unsubscribe(&fake_id).await;
        assert!(result.is_err(), "取消不存在的订阅应该失败");

        // 取消空ID订阅
        let result = bus.unsubscribe("").await;
        assert!(result.is_err(), "取消空ID订阅应该失败");

        // 取消无效格式ID订阅
        let result = bus.unsubscribe("invalid-format-id").await;
        assert!(result.is_err(), "取消无效格式ID订阅应该失败");
    }

    /// 测试 MessageBus::subscriber_count 方法
    #[tokio::test]
    async fn test_subscriber_count_all_scenarios() {
        let bus = TestHelper::create_test_bus().await;

        // 测试不存在主题的订阅者数量
        let count = bus.subscriber_count("nonexistent.topic").await.unwrap();
        assert_eq!(count, 0, "不存在的主题订阅者数量应该为0");

        // 创建多个订阅者
        let topic = "test.count";
        let _sub1 = bus.subscribe(topic).await.unwrap();
        let _sub2 = bus.subscribe(topic).await.unwrap();
        let _sub3 = bus.subscribe("other.topic").await.unwrap();

        let count = bus.subscriber_count(topic).await.unwrap();
        assert_eq!(count, 2, "应该有2个订阅者订阅了指定主题");

        // 测试通配符主题的订阅者数量
        let wildcard_count = bus.subscriber_count("test.*").await.unwrap();
        assert_eq!(wildcard_count, 2, "通配符主题应该匹配2个订阅者");

        // 测试空主题
        let result = bus.subscriber_count("").await;
        assert!(result.is_err(), "空主题查询应该失败");
    }

    /// 测试 MessageBus::list_topics 方法
    #[tokio::test]
    async fn test_list_topics_comprehensive() {
        let bus = TestHelper::create_test_bus().await;

        // 初始状态应该没有主题
        let topics = bus.list_topics().await.unwrap();
        assert!(topics.is_empty(), "初始状态应该没有主题");

        // 创建订阅者会创建主题
        let _sub1 = bus.subscribe("topic1.events").await.unwrap();
        let _sub2 = bus.subscribe("topic2.commands").await.unwrap();
        let _sub3 = bus.subscribe("topic1.events").await.unwrap(); // 重复主题

        let topics = bus.list_topics().await.unwrap();
        assert_eq!(topics.len(), 2, "应该有2个不同的主题");
        assert!(topics.contains(&"topic1.events".to_string()));
        assert!(topics.contains(&"topic2.commands".to_string()));

        // 发布消息也会创建主题
        let message = TestHelper::create_test_message("topic3.notifications", json!({}));
        bus.publish(message).await.unwrap();

        let topics = bus.list_topics().await.unwrap();
        assert_eq!(topics.len(), 3, "发布消息后应该有3个主题");
    }

    /// 测试 MessageBus::has_subscribers 方法
    #[tokio::test]
    async fn test_has_subscribers_all_cases() {
        let bus = TestHelper::create_test_bus().await;

        // 测试没有订阅者的主题
        let has_subs = bus.has_subscribers("empty.topic").await.unwrap();
        assert!(!has_subs, "没有订阅者的主题应该返回false");

        // 创建订阅者
        let topic = "test.subscribers";
        let _subscriber = bus.subscribe(topic).await.unwrap();

        let has_subs = bus.has_subscribers(topic).await.unwrap();
        assert!(has_subs, "有订阅者的主题应该返回true");

        // 测试通配符匹配
        let has_subs = bus.has_subscribers("test.*").await.unwrap();
        assert!(has_subs, "通配符匹配应该找到订阅者");

        // 测试不匹配的主题
        let has_subs = bus.has_subscribers("other.topic").await.unwrap();
        assert!(!has_subs, "不匹配的主题应该返回false");

        // 测试空主题
        let result = bus.has_subscribers("").await;
        assert!(result.is_err(), "空主题查询应该失败");
    }

    /// 测试 MessageBus::close 方法
    #[tokio::test]
    async fn test_close_comprehensive() {
        let bus = TestHelper::create_test_bus().await;

        // 创建一些订阅者和发布一些消息
        let _subscriber = bus.subscribe("test.close").await.unwrap();
        let message = TestHelper::create_test_message("test.close", json!({}));
        bus.publish(message).await.unwrap();

        // 正常关闭
        let result = bus.close().await;
        assert!(result.is_ok(), "正常关闭应该成功");

        // 验证关闭后的状态
        let message = TestHelper::create_test_message("test.after.close", json!({}));
        let result = bus.publish(message).await;
        assert!(result.is_err(), "关闭后发布应该失败");

        let result = bus.subscribe("test.after.close").await;
        assert!(result.is_err(), "关闭后订阅应该失败");

        // 重复关闭
        let result = bus.close().await;
        assert!(result.is_err(), "重复关闭应该失败");
    }

    /// 测试 MessageBus::stats 方法
    #[tokio::test]
    async fn test_stats_comprehensive() {
        let bus = TestHelper::create_test_bus().await;

        // 初始统计
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 0);
        assert_eq!(stats.active_subscribers, 0);

        // 创建订阅者
        let _sub1 = bus.subscribe("stats.test1").await.unwrap();
        let _sub2 = bus.subscribe("stats.test2").await.unwrap();

        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 2);

        // 发布消息
        for i in 0..5 {
            let message = TestHelper::create_test_message(
                &format!("stats.test{}", i % 2 + 1),
                json!({"count": i})
            );
            bus.publish(message).await.unwrap();
        }

        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 5);
        assert_eq!(stats.active_subscribers, 2);
        
        // 验证其他统计字段
        assert!(stats.total_received >= 0);
        assert!(stats.total_failed >= 0);
    }
}

mod publisher_interface_tests {
    use super::*;

    /// 测试 MessageBus::publish 方法作为 Publisher 接口
    #[tokio::test]
    async fn test_messagebus_as_publisher() {
        let bus = TestHelper::create_test_bus().await;

        // 测试基本发布
        let message = TestHelper::create_test_message("publisher.test", json!({"data": "test"}));
        let result = bus.publish(message).await;
        assert!(result.is_ok(), "基本发布应该成功");

        // 测试不同优先级消息
        let priorities = vec![
            MessagePriority::Low,
            MessagePriority::Normal,
            MessagePriority::High,
            MessagePriority::Critical,
        ];

        for priority in priorities {
            let message = Message::new("priority.test", json!({"priority": format!("{:?}", priority)}));
            let mut metadata = message.metadata().clone();
            metadata.attributes.priority = priority;
            let message_with_priority = message.with_metadata(metadata);
            
            let result = bus.publish(message_with_priority).await;
            assert!(result.is_ok(), "优先级 {:?} 消息发布应该成功", priority);
        }
    }

    /// 测试批量发布（手动实现）
    #[tokio::test]
    async fn test_manual_batch_publish() {
        let bus = TestHelper::create_test_bus().await;

        // 手动实现批量发布
        let messages: Vec<Message> = (0..10).map(|i| {
            TestHelper::create_test_message(
                &format!("batch.test.{}", i),
                json!({"id": i, "batch": true})
            )
        }).collect();

        // 一个一个发布
        for message in messages {
            let result = bus.publish(message).await;
            assert!(result.is_ok(), "批量发布中的消息应该成功");
        }

        // 验证统计
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 10);
    }

    /// 测试大批量发布
    #[tokio::test]
    async fn test_large_batch_publish() {
        let bus = TestHelper::create_test_bus().await;

        // 创建大批量消息
        let large_batch_size = 1000;
        for i in 0..large_batch_size {
            let message = TestHelper::create_test_message(
                "batch.large",
                json!({"id": i})
            );
            let result = bus.publish(message).await;
            assert!(result.is_ok(), "大批量发布第 {} 条消息应该成功", i);
        }

        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, large_batch_size);
    }

    /// 测试发布确认机制（通过订阅者验证）
    #[tokio::test]
    async fn test_publish_confirmation_mechanism() {
        let bus = TestHelper::create_test_bus().await;

        // 创建订阅者确保消息被接收
        let subscriber = bus.subscribe("confirm.test").await.unwrap();

        // 发布消息
        let message = TestHelper::create_test_message("confirm.test", json!({"confirmed": true}));
        let result = bus.publish(message).await;
        assert!(result.is_ok(), "发布应该成功");

        // 验证消息被接收
        let received = subscriber.receive_timeout(1000).await.unwrap();
        assert!(received.is_some(), "消息应该被订阅者接收");

        // 测试没有订阅者的情况
        let message = TestHelper::create_test_message("no.subscriber", json!({}));
        let result = bus.publish(message).await;
        assert!(result.is_ok(), "没有订阅者时发布也应该成功");
    }

    /// 测试发布者关闭功能（通过消息总线关闭实现）
    #[tokio::test]
    async fn test_publisher_close_functionality() {
        let bus = TestHelper::create_test_bus().await;

        // 正常发布
        let message = TestHelper::create_test_message("before.close", json!({}));
        let result = bus.publish(message).await;
        assert!(result.is_ok(), "关闭前发布应该成功");

        // 关闭消息总线（代替关闭发布者）
        let result = bus.close().await;
        assert!(result.is_ok(), "关闭消息总线应该成功");

        // 关闭后发布
        let message = TestHelper::create_test_message("after.close", json!({}));
        let result = bus.publish(message).await;
        assert!(result.is_err(), "关闭后发布应该失败");
    }
}

mod subscriber_interface_tests {
    use super::*;

    /// 测试 Subscriber::try_receive 方法
    #[tokio::test]
    async fn test_subscriber_try_receive_comprehensive() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("try_receive.test").await.unwrap();

        // 没有消息时应该返回 None
        let result = subscriber.try_receive().await.unwrap();
        assert!(result.is_none(), "没有消息时应该返回None");

        // 发布消息后应该能接收到
        let test_message = TestHelper::create_test_message(
            "try_receive.test",
            json!({"data": "test_try_receive"})
        );
        bus.publish(test_message).await.unwrap();

        let result = subscriber.try_receive().await.unwrap();
        assert!(result.is_some(), "有消息时应该返回Some");
        
        let received = result.unwrap();
        assert_eq!(received.topic(), "try_receive.test");
        assert_eq!(received.payload()["data"], "test_try_receive");

        // 再次尝试接收应该没有消息
        let result = subscriber.try_receive().await.unwrap();
        assert!(result.is_none(), "消息消费后应该没有更多消息");
    }

    /// 测试 Subscriber::receive 方法
    #[tokio::test]
    async fn test_subscriber_receive_comprehensive() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("receive.test").await.unwrap();

        // 在后台发布消息
        let bus_clone = bus.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            let message = TestHelper::create_test_message(
                "receive.test",
                json!({"delayed": "message"})
            );
            bus_clone.publish(message).await.unwrap();
        });

        // 阻塞接收消息
        let result = subscriber.receive().await.unwrap();
        assert!(result.is_some(), "应该接收到延迟发布的消息");
        
        let received = result.unwrap();
        assert_eq!(received.payload()["delayed"], "message");
    }

    /// 测试 Subscriber::receive_timeout 方法
    #[tokio::test]
    async fn test_subscriber_receive_timeout_all_cases() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("timeout.test").await.unwrap();

        // 测试超时情况
        let result = subscriber.receive_timeout(100).await.unwrap();
        assert!(result.is_none(), "超时应该返回None");

        // 测试在超时前接收到消息
        let message = TestHelper::create_test_message(
            "timeout.test",
            json!({"before_timeout": true})
        );
        bus.publish(message).await.unwrap();

        let result = subscriber.receive_timeout(1000).await.unwrap();
        assert!(result.is_some(), "超时前应该接收到消息");
        
        let received = result.unwrap();
        assert_eq!(received.payload()["before_timeout"], true);

        // 测试零超时
        let result = subscriber.receive_timeout(0).await.unwrap();
        assert!(result.is_none(), "零超时应该立即返回None");

        // 测试长超时
        let start_time = std::time::Instant::now();
        let result = subscriber.receive_timeout(500).await.unwrap();
        let elapsed = start_time.elapsed();
        
        assert!(result.is_none(), "长超时没有消息应该返回None");
        assert!(elapsed.as_millis() >= 450, "应该等待接近超时时间");
    }

    /// 测试 Subscriber::ack 方法
    #[tokio::test]
    async fn test_subscriber_ack_comprehensive() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("ack.test").await.unwrap();

        // 发布消息
        let message = TestHelper::create_test_message(
            "ack.test",
            json!({"ack_test": true})
        );
        bus.publish(message).await.unwrap();

        // 接收消息
        let received = subscriber.receive_timeout(1000).await.unwrap().unwrap();

        // 确认消息
        let result = subscriber.ack(&received).await;
        assert!(result.is_ok(), "确认消息应该成功");

        // 确认无效消息
        let invalid_message = TestHelper::create_test_message(
            "invalid.topic",
            json!({"invalid": true})
        );
        let _result = subscriber.ack(&invalid_message).await;
        // 确认无效消息的行为取决于实现
    }

    /// 测试 Subscriber::nack 方法
    #[tokio::test]
    async fn test_subscriber_nack_comprehensive() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("nack.test").await.unwrap();

        // 发布消息
        let message = TestHelper::create_test_message(
            "nack.test",
            json!({"nack_test": true})
        );
        bus.publish(message).await.unwrap();

        // 接收消息
        let received = subscriber.receive_timeout(1000).await.unwrap().unwrap();

        // 拒绝消息并重新入队
        let result = subscriber.nack(&received, true).await;
        assert!(result.is_ok(), "拒绝并重新入队应该成功");

        // 拒绝消息但不重新入队
        let message2 = TestHelper::create_test_message(
            "nack.test",
            json!({"nack_test": 2})
        );
        bus.publish(message2).await.unwrap();
        
        let received2 = subscriber.receive_timeout(1000).await.unwrap().unwrap();
        let result = subscriber.nack(&received2, false).await;
        assert!(result.is_ok(), "拒绝不重新入队应该成功");
    }

    /// 测试 Subscriber::subscription_id 方法
    #[tokio::test]
    async fn test_subscriber_subscription_id() {
        let bus = TestHelper::create_test_bus().await;
        
        // 创建多个订阅者
        let sub1 = bus.subscribe("id.test").await.unwrap();
        let sub2 = bus.subscribe("id.test").await.unwrap();
        let sub3 = bus.subscribe("other.test").await.unwrap();

        // 每个订阅者应该有唯一的ID
        let id1 = sub1.subscription_id();
        let id2 = sub2.subscription_id();
        let id3 = sub3.subscription_id();

        assert!(!id1.is_empty(), "订阅ID不应该为空");
        assert!(!id2.is_empty(), "订阅ID不应该为空");
        assert!(!id3.is_empty(), "订阅ID不应该为空");

        assert_ne!(id1, id2, "不同订阅者应该有不同的ID");
        assert_ne!(id1, id3, "不同订阅者应该有不同的ID");
        assert_ne!(id2, id3, "不同订阅者应该有不同的ID");

        // ID应该是有效的UUID格式
        assert!(Uuid::parse_str(id1).is_ok(), "订阅ID应该是有效的UUID");
        assert!(Uuid::parse_str(id2).is_ok(), "订阅ID应该是有效的UUID");
        assert!(Uuid::parse_str(id3).is_ok(), "订阅ID应该是有效的UUID");
    }

    /// 测试 Subscriber::pattern 方法
    #[tokio::test]
    async fn test_subscriber_pattern() {
        let bus = TestHelper::create_test_bus().await;
        
        let patterns = vec![
            "exact.pattern",
            "wildcard.*",
            "multilevel.**",
            "complex.*.pattern.*",
        ];

        for pattern in patterns {
            let subscriber = bus.subscribe(pattern).await.unwrap();
            assert_eq!(subscriber.pattern(), pattern, "返回的模式应该与订阅模式一致");
        }
    }

    /// 测试 Subscriber::is_active 方法
    #[tokio::test]
    async fn test_subscriber_is_active() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("active.test").await.unwrap();

        // 新创建的订阅者应该是活跃的
        assert!(subscriber.is_active(), "新创建的订阅者应该是活跃的");

        // 关闭订阅者后应该不活跃
        subscriber.close().await.unwrap();
        assert!(!subscriber.is_active(), "关闭后的订阅者应该不活跃");
    }

    /// 测试 Subscriber::close 方法
    #[tokio::test]
    async fn test_subscriber_close() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("close.test").await.unwrap();

        // 关闭前应该可以接收消息
        let message = TestHelper::create_test_message(
            "close.test",
            json!({"before_close": true})
        );
        bus.publish(message).await.unwrap();
        
        let received = subscriber.receive_timeout(1000).await.unwrap();
        assert!(received.is_some(), "关闭前应该能接收消息");

        // 关闭订阅者
        let result = subscriber.close().await;
        assert!(result.is_ok(), "关闭订阅者应该成功");

        // 关闭后尝试接收消息
        let result = subscriber.try_receive().await;
        assert!(result.is_err(), "关闭后接收消息应该失败");

        // 重复关闭
        let result = subscriber.close().await;
        assert!(result.is_err(), "重复关闭应该失败");
    }

    /// 测试 Subscriber::stats 方法
    #[tokio::test]
    async fn test_subscriber_stats() {
        let bus = TestHelper::create_test_bus().await;
        let subscriber = bus.subscribe("stats.test").await.unwrap();

        // 初始统计
        let stats = subscriber.stats().await.unwrap();
        assert_eq!(stats.total_received, 0);
        assert_eq!(stats.total_acked, 0);

        // 发布和接收消息
        for i in 0..5 {
            let message = TestHelper::create_test_message(
                "stats.test",
                json!({"count": i})
            );
            bus.publish(message).await.unwrap();
            
            let received = subscriber.receive_timeout(1000).await.unwrap();
            assert!(received.is_some());
            
            let msg = received.unwrap();
            subscriber.ack(&msg).await.unwrap();
        }

        // 验证统计
        let stats = subscriber.stats().await.unwrap();
        assert_eq!(stats.total_received, 5);
        assert_eq!(stats.total_acked, 5);
    }
}